package drds.data_migrate.task_group;


import drds.common.Author;
import drds.common.Else;
import drds.data_migrate.Authors;
import drds.data_migrate.task_group.task._runnable_.AbstractRunnable;
import drds.data_migrate.task_group.task._runnable_.ReadRunnable;
import drds.data_migrate.task_group.task._runnable_.WriteRunnable;
import drds.data_migrate.task_group.task.data_exchange.exchanger.Produce;
import drds.data_migrate.task_group.task.data_exchange.exchanger.RecordExchanger;
import drds.data_migrate.task_group.task.data_exchange.exchanger.RecordQueue;
import drds.data_migrate.vo.State;
import drds.data_migrate.vo.Status;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;

/**
 * TaskExecutor是一个完整task的执行器,其中包括1：1的reader和writer
 */
@Author(name = Authors.LI_YANG)
class TaskExecutor
{
    @Setter
    @Getter
    private TaskGroupContainer taskGroupContainer;
    @Setter
    @Getter
    private TaskConfig taskConfig;
    @Setter
    @Getter
    private int taskId;
    @Setter
    @Getter
    private int attemptCount;

    //

    @Setter
    @Getter
    private RecordQueue recordQueue;
    @Setter
    @Getter
    private ReadRunnable readRunnable;
    @Setter
    @Getter
    private WriteRunnable writeRunnable;
    @Setter
    @Getter
    private Thread readeThread;
    @Setter
    @Getter
    private Thread writeThread;

    @Setter
    @Getter
    private State state;

    public TaskExecutor(TaskGroupContainer taskGroupContainer, TaskConfig taskConfig, int attemptCount)
    {
        this.taskGroupContainer = taskGroupContainer;

        this.taskConfig = taskConfig;

        this.taskId = this.taskConfig.getTaskId();
        this.attemptCount = attemptCount;

        /**
         * 由taskId得到该taskExecutor的Communication
         * 要传给readerRunner和writerRunner，同时要传给channel作统计用
         */
        this.state = null;
        this.recordQueue = new RecordQueue(taskConfig);
        this.recordQueue.setState(this.state);
        //
        writeRunnable = (WriteRunnable) newRunnable(TaskType.write);
        this.writeThread = new Thread(writeRunnable, String.format("%d-%d-%d-write", taskGroupContainer.jobId, taskGroupContainer.taskGroupId, this.taskId));
        //
        readRunnable = (ReadRunnable) newRunnable(TaskType.reade);
        this.readeThread = new Thread(readRunnable, String.format("%d-%d-%d-reade", taskGroupContainer.jobId, taskGroupContainer.taskGroupId, this.taskId));

    }

    public AbstractRunnable newRunnable(TaskType taskType)
    {
        AbstractRunnable abstractRunnable = null;


        switch (taskType)
        {
            case reade:
                abstractRunnable = new ReadRunnable();//
                Produce produce = new RecordExchanger(this.recordQueue);

                ((ReadRunnable) abstractRunnable).setProduce(produce);
                break;
            case write:
                abstractRunnable = new WriteRunnable();//
                ((WriteRunnable) abstractRunnable).setConsume(new RecordExchanger(this.recordQueue));

                break;
            default:
                throw new Else();
        }

        abstractRunnable.setTaskGroupId(taskGroupContainer.taskGroupId);
        abstractRunnable.setTaskId(this.taskId);
        abstractRunnable.setStateMessage(this.state);
        return abstractRunnable;
    }

    //
    public void doStart()
    {
        this.writeThread.start();
        // reader没有起来，writer不可能结束
        if (!this.writeThread.isAlive() || this.state.getStatus() == Status.failed)
        {
            throw new RuntimeException();
        }
        this.readeThread.start();
        // 这里reader可能很快结束
        if (!this.readeThread.isAlive() && this.state.getStatus() == Status.failed)
        {
            // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
            throw new RuntimeException();
        }

    }

    // 检查任务是否结束
    public boolean isTaskFinished()
    {
        // 如果reader 或 writer没有完成工作，那么直接返回工作没有完成
        if (readeThread.isAlive() || writeThread.isAlive())
        {
            return false;
        }
        if (state == null || !state.isFinished())
        {
            return false;
        }
        return true;
    }

    //
    public void shutdown()
    {
        writeRunnable.shutdown();
        readRunnable.shutdown();
        if (writeThread.isAlive())
        {
            writeThread.interrupt();
        }
        if (readeThread.isAlive())
        {
            readeThread.interrupt();
        }
    }

    public boolean isShutdown()
    {
        return !readeThread.isAlive() && !writeThread.isAlive();
    }
}
